package com.example.streamdemo;

import org.springframework.boot.CommandLineRunner;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.function.Consumer;

/**
 * @description:
 * @author: 吴佳伟
 * @date: 2021/5/17
 */
@Component
public class IntegrationFunction implements CommandLineRunner {


    private final StreamBridge stream;

    public IntegrationFunction(StreamBridge stream) {
        this.stream = stream;
    }


    @Bean
        //这里接收rabbitmq的条件是参数为Consumer 并且 方法名和supplier方法名相同
        //这里的返回值是一个匿名函数 返回类型是consumer 类型和提供者的类型一致
        //supplier发送的exchange是 send-in-0 这里只需要用send方法名即可
    Consumer<String> run() {
        return str -> System.out.println("我是消费者，我收到了消息：" + str);
    }


    @Override
    public void run(String... args) {
        boolean send = stream.send("run-in-0", "12345");
        System.out.println("send message of 12345 ");
    }

    public void threadRun(){
        SimpleAsyncTaskExecutor threadPoolExecutor = new SimpleAsyncTaskExecutor("demo-1");
        threadPoolExecutor.submit((Runnable) () -> {
            while (true) {
                boolean send = stream.send("run-in-0", "12345");
                System.out.println("send message of 12345 ");
            }
        });
    }
}
